/*
 * Licensed to the Tai Ping Jin Ke
 *
 * Copyright (c) 2022 .
 * All rights reserved.
 * 项目名称：太平金科-基础组件封装层-会话管理-reactor-缺省实现
 * 版权说明：本软件属太平金融科技服务(上海)有限公司所有，在未获得太平金融科技服务(上海)有限公司正式授权情况下，任何企业和个人，不能获取、阅读、安装、传播本软件涉及的任何受知识产权保护的内容。
 */
package com.taipingframework.boot.reactor.session;

import com.fasterxml.jackson.core.type.TypeReference;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.taipingframework.boot.cache.core.ReactiveCacheHandler;
import com.taipingframework.boot.cache.redis.DynamicCacheKeyBuilder;
import com.taipingframework.boot.cache.redis.StandardCacheKey;
import com.taipingframework.boot.session.ReactorSessionManager;
import com.taipingframework.boot.session.constant.SessionCachePathEnum;
import com.taipingframework.boot.session.constant.SessionConstant;
import com.taipingframework.boot.session.core.ReactiveSessionHandler;
import com.taipingframework.boot.session.support.ClientDetails;
import com.taipingframework.boot.web.response.status.ApiStatusEnum;
import com.taipingframework.boot.web.response.status.ExceptionStatusEnum;
import com.taipingframework.utility.constant.ApplicationConstant;
import com.taipingframework.utility.exception.SystemInterrupterException;
import com.taipingframework.utility.exception.UtilityException;
import com.taipingframework.utility.extend.ReactorHelper;
import com.taipingframework.utility.extend.StringHelper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.lang.NonNull;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.WebSession;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Arrays;
import java.util.Collections;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.function.Function;

@Slf4j
public class DefaultSessionHandler extends OriginalSessionHandler implements ReactiveSessionHandler {

    public DefaultSessionHandler(ReactiveCacheHandler reactiveCacheHandler, ObjectMapper objectMapper) {
        super(reactiveCacheHandler, objectMapper);
    }

    /**
     * 创建Session [登录]
     * <p>
     * 特别说明：
     * `在会话有效期内使用了错误的身份凭证`这个错误消息针对的是在权限管理模块开放了白名单的登录接口，
     * 向前端抛出这个错误消息的目的是为了限制客户端使用相同的账号重复创建多个不同的会话对象！
     * <p>
     * 登录接口的特殊性：
     * 登录接口加入白名单后，那么在调用该接口时不要求提供token参数，
     * 但是，如果客户端已经持有token，则在多次重复调用登录接口时，应该提供入参token；
     * 否则，抛出错误消息：`在会话有效期内使用了错误的身份凭证`。
     * <p>
     * 个人理解：
     * 正常情况下，在权限管理模块应该只能允许“登录接口”被加入白名单，
     * 而如果存在其它被加入到白名单的接口，只能被允许提供读取数据的权限，禁止写入数据！
     * <p>
     * 常规方案：
     * 排除登录接口，针对其它被加入到白名单的接口，正确做法是作为open-api面向网络开发，在调用链中不会经过权限控制的一系列filter。
     * 另外，针对网络接口，在被客户端重复调用时，应该充分考虑数据的幂等性问题并给予解决，避免数据冗余！
     */
    @Override
    public Mono<WebSession> createSession(@NonNull String loginName) {
        // 读取会话对象
        Function<ClientDetails, Mono<WebSession>> readWebSessionFunc = (clientDetails) ->
                ReactorHelper.getExchange().flatMap(exchange -> super.readWebSession(exchange)
                        .doFirst(() -> log.debug("终端显示层【{}】会话对象未失效，不会重复创建会话对象", clientDetails.getClientIpAddress()))
                        .doOnError(e -> {
                            // 读取缓存服务器的会话数据
                            ReactorSessionManager.readCurrentTerminalSession(loginName, clientDetails.getClientIpAddress(), clientDetails.getClientUserProfile())
                                    .subscribe(realSessionId -> {
                                        // 拿到缓存中的会话对象标识，打印异常日志
                                        log.error("终端显示层【{}】会话对象在服务器已经存在，但是获取失败。关键参数：(1)请求头携带的会话标识ID是：{}；(2)实际应该是：{}",
                                                clientDetails.getClientIpAddress(), clientDetails.getRequestHeaderToken(), realSessionId);
                                    });
                        })
                ).onErrorResume(ignored -> {
                    // 重新包装异常消息，给出符合应用场景的异常报文
                    return Mono.error(new SystemInterrupterException(ExceptionStatusEnum.INCORRECT_TOKEN_FROM_CLIENT));
                });

        // 创建会话对象
        Function<ClientDetails, Mono<WebSession>> createWebSessionFunc = (clientDetails) ->
                ReactorHelper.getExchange().flatMap(exchange -> super.createWebSession(exchange)
                        .doOnNext(webSession -> {
                            // 会话对象创建成功后，将登陆名与会话对象绑定
                            webSession.getAttributes().putIfAbsent(SessionConstant.SESSION_LOGIN_KEY, loginName);
                            // 会话对象创建成功后，将用户空间与会话对象绑定
                            webSession.getAttributes().putIfAbsent(SessionConstant.CLIENT_USER_PROFILE_KEY, clientDetails.getClientUserProfile());
                            // 记录当前会话对象映射的登录名【在spring-webflux会话管理模块DefaultSessionHandler.java中退出登录成功时获取】
                            exchange.getAttributes().putIfAbsent(SessionConstant.SESSION_LOGIN_KEY, loginName);

                            // 更新缓存的value
                            ReactorSessionManager.saveOrUpdateCurrentTerminalSession(clientDetails, webSession.getId()).subscribe();
                        })
                        .doOnSuccess(webSession -> log.debug("将会话标识【{}】与终端显示层所在设备的真实IP【{}】进行绑定，并将该映射关系保存在redis缓存中",
                                webSession.getId(), clientDetails.getClientIpAddress()))
                        .doOnError(e -> {
                            // 更优雅地书写思虑周全的代码往往是很困难的，但是，java中的异常就像幽灵似的神出鬼没，它还没出现但却在暗处找机会等着给你惊喜而已！
                            // 如果会话对象创建成功但在执行后续的其它操作时出现了异常，那么这个会话对象的数据就是不健全的，这类脏数据很令人头疼。
                            // 此处我试图通过硬编码的方式进行数据回滚！
                            Mono.just(Optional.ofNullable((String) exchange.getAttribute(authTokenKey)).orElse(StringUtils.EMPTY))
                                    .flatMap(sessionId -> {
                                        if (StringUtils.isNotEmpty(sessionId)) {
                                            // 创建会话失败，删除已经创建的临时会话凭据
                                            return ReactorSessionManager.deleteCurrentTerminalSession(loginName,
                                                    clientDetails.getClientIpAddress(),
                                                    clientDetails.getClientUserProfile(),
                                                    sessionId
                                            ).then(cacheHandler.delete(sessionId, SessionCachePathEnum.SESSION_DATA_PATH));
                                        }
                                        return Mono.just(false);
                                    }).subscribe(null, null, () -> log.error("会话对象创建失败执行数据回滚操作"));
                        })
                ).onErrorResume(e -> Mono.error(new UtilityException(e)));

        return ReactorSessionManager.readClientDetails(loginName)
                //.subscribeOn(Schedulers.single())
                .zipWhen(ReactorSessionManager::validateClientInfo, (clientDetails, validResult) -> {
                    if (validResult) {
                        return readWebSessionFunc.apply(clientDetails);
                    } else {
                        return createWebSessionFunc.apply(clientDetails);
                    }
                })
                .flatMap(webSessionMono -> webSessionMono)
                //.publishOn(Schedulers.boundedElastic())
                .onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

    /**
     * 移除Session [注销]
     * <p>
     * 退出登录接口的调用受权限控制模块的管理，调用时要求提供token；
     * 无效的token不允许被访问！
     * <p>
     * 登录接口是被加入白名单的，所以有N多种创建本地会话的场景需要枚举并验证；
     * 除了登录接口，其它接口都是需要经过权限管理模块的统一验证才能被调用！
     */
    @Override
    public Mono<Boolean> removeSession() {
        return ReactorHelper.getExchange()
                .zipWhen(ServerWebExchange::getSession, ((exchange, webSession) -> {
                    // 获取当前会话映射的登录名
                    String loginName = webSession.getAttributeOrDefault(SessionConstant.SESSION_LOGIN_KEY, StringUtils.EMPTY);
                    // 获取当前会话映射的远程客户端设备标识
                    String clientIpAddress = webSession.getAttributeOrDefault(ApplicationConstant.REMOTE_CLIENT_IP_KEY, StringUtils.EMPTY);
                    // 获取当前会话映射的远程客户端用户空间
                    String flagManagedProfile = webSession.getAttributeOrDefault(SessionConstant.CLIENT_USER_PROFILE_KEY, StringUtils.EMPTY);

                    // 如果此处可以注销成功，则上面的这3个属性是可以信任的
                    return super.removeWebSession(exchange)
                            .flatMap(actionResult -> {
                                if (actionResult) {
                                    // 获取请求头中的token
                                    String accessToken = Optional.ofNullable(exchange.getRequest().getHeaders().getFirst(authTokenKey))
                                            .orElse(StringUtils.EMPTY);
                                    // 同步移除缓存中的验证数据
                                    return ReactorSessionManager.deleteCurrentTerminalSession(loginName, clientIpAddress, flagManagedProfile, accessToken)
                                            .doOnSuccess(voidMono -> log.debug("会话标识【{}】与终端显示层【{}】的映射关系已从redis缓存中清除", accessToken, clientIpAddress))
                                            .thenReturn(true);
                                }
                                return Mono.just(false);
                            });
                }))
                .flatMap(actionResultMono -> actionResultMono)
                .onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

    /**
     * 暴力移除会话[强制注销/强制下线]
     * <p>
     * 在同一台设备上，可用的终端是有限的，用户切换账号时后端需要主动负责将设备IP+用户空间映射的用户会话给注销掉，
     * 而不是让别人手动退出后重新输入账号、密码后才可以登录。
     * <p>
     * 该方法不应被调用，而应该由前端通过参数`conflictSessionId`主动调用退出登录的接口，然后才能再次调用登录的接口！
     */
    @Override
    @Deprecated
    public Mono<Boolean> removeSessionByForce(@NonNull String conflictSessionId) {
        return cacheHandler.getHash(conflictSessionId, SessionCachePathEnum.SESSION_DATA_PATH)
                .flatMap(sessionMapResult -> {
                    String loginName = sessionMapResult.getOrDefault(
                            ReactorSessionManager.buildAttributeCacheKey(SessionConstant.SESSION_LOGIN_KEY), StringUtils.EMPTY);
                    String clientIpAddress = sessionMapResult.getOrDefault(
                            ReactorSessionManager.buildAttributeCacheKey(ApplicationConstant.REMOTE_CLIENT_IP_KEY), StringUtils.EMPTY);
                    String flagManagedProfile = sessionMapResult.getOrDefault(
                            ReactorSessionManager.buildAttributeCacheKey(SessionConstant.CLIENT_USER_PROFILE_KEY), StringUtils.EMPTY);

                    ClientDetails clientDetails = new ClientDetails();
                    clientDetails.setClientLoginName(StringHelper.trim(loginName, "\""));
                    clientDetails.setClientIpAddress(StringHelper.trim(clientIpAddress, "\""));
                    clientDetails.setClientUserProfile(StringHelper.trim(flagManagedProfile, "\""));
                    clientDetails.setClientPresentToken(conflictSessionId);
                    return Mono.just(clientDetails);
                })
                .flatMap(clientDetails -> {
                    // 移除会话数据
                    return cacheHandler.delete(clientDetails.getClientPresentToken(), SessionCachePathEnum.SESSION_DATA_PATH)
                            .then(ReactorSessionManager.deleteCurrentTerminalSession(clientDetails.getClientLoginName(),
                                    clientDetails.getClientIpAddress(),
                                    clientDetails.getClientUserProfile(),
                                    clientDetails.getClientPresentToken()))
                            .thenReturn(true);
                })
                .onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

    /**
     * 判断当前登录操作是否与已有会话对象冲突
     *
     * @return 返回已经存在的会话对象标识
     */
    @Override
    public Mono<String> checkSessionConflict(@NonNull String loginName) {
        return ReactorSessionManager.readClientUserProfile()
                .zipWith(ReactorHelper.getExchange(), (flagManagedProfile, exchange) -> {
                    String clientIpAddress = exchange.getAttributeOrDefault(ApplicationConstant.REMOTE_CLIENT_IP_KEY, StringUtils.EMPTY);
                    return cacheHandler.getHashValue(clientIpAddress, SessionCachePathEnum.SESSION_PROFILE_PATH, flagManagedProfile)
                            .defaultIfEmpty(StringUtils.EMPTY)
                            .flatMap(signPrincipal -> {
                                if (StringUtils.isEmpty(signPrincipal)) {
                                    return Mono.just(StringUtils.EMPTY);
                                }
                                if (loginName.equalsIgnoreCase(signPrincipal)) {
                                    return Mono.just(StringUtils.EMPTY);
                                }

                                String uniqueTokenKey = ReactorSessionManager.buildUniqueTokenKey(signPrincipal, clientIpAddress, flagManagedProfile);
                                return cacheHandler.getValue(uniqueTokenKey, SessionCachePathEnum.SESSION_UNEXPIRED_PATH)
                                        .defaultIfEmpty(StringUtils.EMPTY);
                            });
                })
                .flatMap(tokenMono -> tokenMono)
                .onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

    /**
     * 获取当前会话标识
     * <p>
     * 获取当前登录用户接口的调用受权限控制模块的管理，调用时要求提供token；
     * 无效的token不允许被访问！
     * <p>
     * 登录接口是被加入白名单的，所以有N多种创建本地会话的场景需要枚举并验证；
     * 除了登录接口，其它接口都是需要经过权限管理模块的统一验证才能被调用！
     */
    @Override
    public Mono<WebSession> readSession() {
        return ReactorHelper.getExchange()
                .flatMap(super::readWebSession)
                .onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

    /**
     * 设置属性值
     */
    @Override
    public Mono<Boolean> setAttribute(@NonNull String name, Object value) {
        if (StringUtils.isBlank(name) || Objects.isNull(value)) {
            return Mono.just(false);
        }

        Function<Object, String> serializeValue = (source) -> {
            try {
                String cacheValue;
                if (source instanceof CharSequence) {
                    cacheValue = ((CharSequence) source).toString();
                } else {
                    cacheValue = objectMapper.writeValueAsString(source);
                    cacheValue = objectMapperForRedis.writeValueAsString(JSONUtil.parseObj(cacheValue));
                }
                return cacheValue;
            } catch (Exception e) {
                throw new UtilityException(e);
            }
        };

        return ReactorHelper.getExchange()
                .map(exchange -> exchange.getAttributeOrDefault(authTokenKey, StringUtils.EMPTY))
                .filter(StringUtils::isNotEmpty)
                .switchIfEmpty(this.readSession().map(WebSession::getId))
                .zipWith(Mono.fromSupplier(() -> serializeValue.apply(value)),
                        (tokenValue, cacheValue) -> {
                            String cacheKey = DynamicCacheKeyBuilder.createUniqueCacheKey(tokenValue, SessionCachePathEnum.SESSION_DATA_PATH);
                            String hashName = ReactorSessionManager.buildAttributeCacheKey(name);
                            // 警告：此处不允许再次对会话数据进行刷新，否则有可能导致spring-session过期机制在运行时产生紊乱
                            return cacheHandler.putHash(cacheKey, hashName, cacheValue);
                            // 这个API在执行时，如果更新缓存成功就会重新设置一个有效期。二次封装，请不要干预spring-session会话数据的过期机制，有可能导致未知风险。
                            //return cacheHandler.putHash(tokenValue, SessionCachePathEnum.SESSION_DATA_PATH, hashName, cacheValue);
                        }
                ).flatMap(booleanMono -> booleanMono)
                .doOnError(e -> ReactorHelper.getRequest()
                        .map(request -> request.getHeaders().getFirst(authTokenKey))
                        .doOnSuccess(sessionId -> log.error("设置当前会话对象(session-id：{})的属性 {} 时发生异常", sessionId, name)))
                .onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

    /**
     * 设置属性值
     */
    @Override
    public Mono<Boolean> setAttribute(@NonNull StandardCacheKey cacheKeyEnum, Object value) {
        return setAttribute(cacheKeyEnum.getKey(), value)
                .onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

    /**
     * 移除属性
     */
    @Override
    public Mono<Long> removeAttribute(@NonNull String name) {
        if (StringUtils.isBlank(name)) {
            return Mono.just(0L);
        }

        return ReactorHelper.getExchange()
                .map(exchange -> exchange.getAttributeOrDefault(authTokenKey, StringUtils.EMPTY))
                .filter(StringUtils::isNotEmpty)
                .switchIfEmpty(this.readSession().map(WebSession::getId))
                .flatMap(tokenValue -> {
                    String hashName = ReactorSessionManager.buildAttributeCacheKey(name);
                    return cacheHandler.removeHashEntry(tokenValue, SessionCachePathEnum.SESSION_DATA_PATH, Collections.singletonList(hashName));
                })
                .doOnError(e -> ReactorHelper.getRequest()
                        .map(request -> request.getHeaders().getFirst(authTokenKey))
                        .doOnSuccess(sessionId -> log.error("移除当前会话对象(session-id：{})的属性({})失败", sessionId, name)))
                .onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

    /**
     * 移除属性
     */
    @Override
    public Mono<Long> removeAttribute(@NonNull StandardCacheKey cacheKeyEnum) {
        return removeAttribute(cacheKeyEnum.getKey())
                .onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

    /**
     * 获取属性值
     */
    @Override
    public Mono<String> getAttribute(@NonNull String name) {
        if (StringUtils.isBlank(name)) {
            return Mono.just(StringUtils.EMPTY);
        }

        return ReactorHelper.getExchange()
                .map(exchange -> exchange.getAttributeOrDefault(authTokenKey, StringUtils.EMPTY))
                .filter(StringUtils::isNotEmpty)
                .switchIfEmpty(this.readSession().map(WebSession::getId))
                .flatMap(tokenValue -> {
                    String hashName = ReactorSessionManager.buildAttributeCacheKey(name);
                    return cacheHandler.getHashValue(tokenValue, SessionCachePathEnum.SESSION_DATA_PATH, hashName);
                })
                .flatMap(cacheValue -> {
                    if (JSONUtil.isTypeJSONArray(cacheValue)) {
                        JSONArray jsonArray = JSONUtil.parseArray(cacheValue);
                        String cachedObjectType = String.valueOf(jsonArray.get(0));
                        if (Arrays.asList(JSONObject.class.getName(), JSONArray.class.getName()).contains(cachedObjectType)) {
                            return Mono.fromSupplier(() -> JSONUtil.toJsonStr(jsonArray.get(1)));
                        } else {
                            log.error(String.format("不能处理%s类型的JSON数组", cachedObjectType));
                            throw new SystemInterrupterException(ApiStatusEnum.ERR_A0421);
                        }
                    }
                    return Mono.just(cacheValue);
                })
                .doOnError(e -> ReactorHelper.getRequest()
                        .map(request -> request.getHeaders().getFirst(authTokenKey))
                        .doOnSuccess(sessionId -> log.error("获取当前会话对象(session-id：{})的属性({})失败", sessionId, name)))
                .onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

    /**
     * 获取属性值
     */
    @Override
    public Mono<String> getAttribute(@NonNull StandardCacheKey cacheKeyEnum) {
        return getAttribute(cacheKeyEnum.getKey())
                .onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

    /**
     * 获取属性值
     */
    @Override
    public <T> Mono<T> getAttribute(@NonNull StandardCacheKey cacheKeyEnum, Class<T> clazz) {
        return getAttribute(cacheKeyEnum)
                .map(cacheValue -> {
                    try {
                        return objectMapper.readValue(cacheValue, clazz);
                    } catch (Exception e) {
                        throw new UtilityException(e);
                    }
                });
    }

    /**
     * 获取属性值
     */
    @Override
    public <T> Mono<T> getAttribute(@NonNull StandardCacheKey cacheKeyEnum, TypeReference<T> type) {
        return getAttribute(cacheKeyEnum)
                .map(cacheValue -> {
                    try {
                        return objectMapper.readValue(cacheValue, type);
                    } catch (Exception e) {
                        throw new UtilityException(e);
                    }
                });
    }

    /**
     * 获取所有的属性名列表
     */
    @Override
    public Flux<String> getAttributeNames() {
        return ReactorHelper.getExchange()
                .map(exchange -> exchange.getAttributeOrDefault(authTokenKey, StringUtils.EMPTY))
                .filter(StringUtils::isNotEmpty)
                .switchIfEmpty(this.readSession().map(WebSession::getId))
                // 获取缓存服务器的会话对象数据【优化措施：此处应该返回Set<String>而不必返回Map<String, String>】
                .flatMap(sessionId -> cacheHandler.getHash(sessionId, SessionCachePathEnum.SESSION_DATA_PATH))
                .flatMapIterable(Map::keySet)
                .doOnError(e -> ReactorHelper.getRequest()
                        .map(request -> request.getHeaders().getFirst(authTokenKey))
                        .doOnSuccess(sessionId -> log.error("获取当前会话对象(session-id：{})的属性列表失败", sessionId)))
                .onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

}
